package com.weshare.transform;


import com.weshare.apitest.beans.SensorReading;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.collector.selector.OutputSelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.SplitStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.Collections;

public class TransformTest4_Multi {

    public static void main(String[] args) throws Exception {

        // 创建流处理的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 从文件中读取数据
        String inputPath = "D:\\idea-workspace\\flink-learn\\src\\main\\resources\\sensor.txt";
        DataStream<String> inputStream = env.readTextFile(inputPath);

        // 转换读取的对象
        DataStream<SensorReading> dataStream = inputStream.map(new MapFunction<String, SensorReading>() {
            @Override
            public SensorReading map(String value) throws Exception {
                String[] strings = value.split(",");
                System.out.println(strings);
                return new SensorReading(strings[0], Long.valueOf(strings[1]), Double.parseDouble(strings[2]));

            }
        });

        SplitStream<SensorReading> split = dataStream.split(new OutputSelector<SensorReading>() {
            @Override
            public Iterable<String> select(SensorReading value) {
                return value.getTemperature() > 30 ? Collections.singletonList("HIGH") : Collections.singletonList("LOW");
            }
        });

        DataStream<SensorReading> highStream = split.select("HIGH");
        DataStream<SensorReading> lowStream = split.select("LOW");
        DataStream<SensorReading> allStream = split.select("HIGH","LOW");

        highStream.print("HIGH");
        lowStream.print("LOW");
        allStream.print("ALL");

        // 执行
        env.execute();
    }
}
